package javaVersion.learn.checkPoint;

import akka.stream.Supervision;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.runtime.state.StateBackend;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.io.IOException;
import java.util.concurrent.TimeUnit;

/**
 * @program: myFlink
 * @description: FLink容错checkpoint方法使用
 * @author: WincoMa
 * @create: 2020-07-21 10:40
 **/
public class CheckPointDemo {
    public static void main(String[] args) throws IOException {
        StreamExecutionEnvironment senv = StreamExecutionEnvironment.getExecutionEnvironment();
        CheckpointConfig ck = senv.getCheckpointConfig();

        // TODO: 2020/7/21 设置当作业取消时，状态是否保留 
        //当作业取消时，保留作业的 checkpoint。注意，这种情况下，需要手动清除该作业保留的 checkpoint。
        ck.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //当作业取消时，删除作业的 checkpoint。
        //ck.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION);

        // TODO: 2020/7/21 设置checkpoint的执行模式，最多执行一次或者至少执行一次
        //精确一次执行
        ck.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        //至少执行一次
        //ck.setCheckpointingMode(CheckpointingMode.AT_LEAST_ONCE);

        // TODO: 2020/7/21 设置ck的超时时间
        ck.setCheckpointTimeout(60000);

        // TODO: 2020/7/21 如果在快照制作过程中出现错误，是否让整体任务失败，true/false
        ck.setFailOnCheckpointingErrors(false);

        // TODO: 2020/7/21 设置同一时间有多少个ck可以同时执行
        ck.setMaxConcurrentCheckpoints(1);

        // TODO: 2020/7/21 设置ck的持久化方式（内存，文件系统，RocksDB数据库）
        //注意这里setStateBackend(AbstractStateBackend backend)方法过时
        //解决方案就是将FsStateBackend单独拿出来new，实现以StateBackend为父类的方法。
        StateBackend fsStateBackend = new FsStateBackend("");
        senv.setStateBackend(fsStateBackend);
        // TODO: 2020/7/22 设置重启策略,对批进行设置
//        senv.setRestartStrategy(Supervision.Restart.);
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,
                Time.of(10, TimeUnit.SECONDS)));

    }
}
